package com.foreveross.taskservice.taskcache.redis;

import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;

import com.foreveross.taskservice.common.CacheDataSpace;
import com.foreveross.taskservice.common.TaskStatusEnum;
import com.foreveross.taskservice.common.model.AbstractAction;
import com.foreveross.taskservice.common.model.AbstractTaskModel;
import com.foreveross.taskservice.common.model.airfreight.AirFreightTaskModel;
import com.foreveross.taskservice.infra.cfg.PropertyUtils;
import com.foreveross.taskservice.infra.cfg.taskservice.TaskServicePropertiesLoader;
import com.foreveross.taskservice.taskcache.QueryIndex;
import com.foreveross.taskservice.taskcache.TaskAddQueryIndexApplication;
import com.foreveross.taskservice.taskcache.TaskRollbackConvertorApplication;
import com.foreveross.taskservice.taskcache.redis.core.LockTypeEnum;
import com.foreveross.taskservice.taskcache.redis.core.TaskRedisCallback;
import com.foreveross.taskservice.taskcache.redis.core.TaskRedisTemplate;
import com.foreveross.taskservice.taskcache.redis.core.exception.MultiTransactionFailException;
import com.foreveross.taskservice.taskcache.redis.util.RedisSerializeUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.ibm.db2.jcc.am.j;

/**
 * 抽象类 提供一些公共的key方法和某些统一标准
 * 
 * @author Administrator
 * 
 */
public abstract class AbstractRedisTaskCache {
	protected final int SAVE_OPERATE = 1;
	protected final int UPDATE_OPERATE = 2;
	protected final int DELETE_OPERATE = 3;

	protected static final Logger log = Logger
			.getLogger(AbstractRedisTaskCache.class);

	public final static String DateFormat = "yyyyMMdd";

	/**
	 * 要求实现redis模板
	 * 
	 * @return
	 */
	protected abstract TaskRedisTemplate getTaskRedisTemplate();

	/**
	 * 要求根据模型，实现对应的查询索引 允许返回null
	 * 
	 * @param model
	 * @return
	 */
	protected abstract TaskAddQueryIndexApplication<AbstractTaskModel> getAddQueryIndexApplication(
			AbstractTaskModel model);
	/**
	 * 要求实现回滚兑换接口
	 * @param rollTask
	 * @param cacheTask
	 * @return
	 */
	protected abstract TaskRollbackConvertorApplication<AbstractTaskModel> getTaskRollbackConvertorApplication(
			AbstractTaskModel rollTask,AbstractTaskModel cacheTask);
	
	
	/**
	 * 验证渠道id
	 * 
	 * @param id
	 * @throws Exception
	 */
	protected void validateChannelId(String channelCode) throws Exception {
		if (channelCode == null) {
			throw new Exception("渠道key为空");
		}
	}

	protected byte[][] buildBytess(Collection<byte[]> cs) throws Exception {
		byte[][] result = new byte[cs.size()][];
		int i = 0;
		for (byte[] c : cs) {
			result[i] = c;
			i++;
		}
		return result;
	}

	/**
	 * 指定数据空间，条件key,生成一个sortSet类型的条件set byte[]
	 * 
	 * @param space
	 * @param conditionKey
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildConditionSortSetKeyByte(CacheDataSpace space,
			String conditionKey) throws Exception {
		return RedisSerializeUtil.serializeString(buildConditionSortSetKey(
				space, conditionKey));
	}

	/**
	 * 指定数据空间，条件key,条件值value生成一个Set类型的条件set byte[]
	 * 
	 * @param space
	 * @param conditionKey
	 * @param value
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildConditionSetKeyByte(CacheDataSpace space,
			String conditionKey, String value) throws Exception {
		return RedisSerializeUtil.serializeString(buildConditionSetKey(space,
				conditionKey, value));
	}

	/**
	 * 指定数据空间，查询条件key,查询条件value创建条件setkey CONDITION:SPACE:key:value:SET
	 * 
	 * @param space
	 * @param conditionKey
	 * @param conditionValue
	 * @return
	 * @throws Exception
	 */
	protected String buildConditionSetKey(CacheDataSpace space,
			String conditionKey, String conditionValue) throws Exception {
		return new StringBuffer().append(TaskCacheContext.SPACE_CONDITION)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(space.getSpace())
				.append(TaskCacheContext.SPACE_DELIMITER).append(conditionKey)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(conditionValue)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(TaskCacheContext.SET).toString();
	}

	/**
	 * 指定数据空间、查询条件key、创建条件SortSetKey CONDITION:SPACE:conditionKey:SORTSET
	 * 
	 * @param space
	 * @param conditionKey
	 * @return
	 * @throws Exception
	 */
	protected String buildConditionSortSetKey(CacheDataSpace space,
			String conditionKey) throws Exception {
		return new StringBuffer().append(TaskCacheContext.SPACE_CONDITION)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(space.getSpace())
				.append(TaskCacheContext.SPACE_DELIMITER).append(conditionKey)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(TaskCacheContext.SORTED_SET).toString();
	}

	/**
	 * 根据任务状态枚举分析出所在队列 如 状态 0 等待中 则应该是等待队列
	 * 
	 * @param status
	 * @return
	 */
	protected String getQueueType(TaskStatusEnum status) {
		if (status == TaskStatusEnum.TASK_WATI) {
			return TaskCacheContext.WAIT_QUEUE;
		} else if (status == TaskStatusEnum.TASK_DOING) {
			return TaskCacheContext.DOING_QUEUE;
		} else {
			return TaskCacheContext.FINISH_QUEUE;
		}
	}

	// ===============================================================================================================

	protected boolean logAction(AbstractTaskModel task, AbstractAction action)
			throws Exception {
		return task.addAction(action);
	}

	/**
	 * 根据游离态任务找出内存中的任务
	 * 
	 * @param space
	 * @param models
	 * @param excluseNull
	 *            排除null
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected LinkedHashSet<AbstractTaskModel> findCacheTasksByDissociateTasks(
			LinkedHashSet<AbstractTaskModel> models, boolean excluseNull,
			Jedis j) throws Exception {
		LinkedHashSet<byte[]> idBytes = Sets.newLinkedHashSet();
		for (AbstractTaskModel m : models) {
			idBytes.add(RedisSerializeUtil.serializeString(m.getId()));
		}
		return findCacheTasks(idBytes, excluseNull, j);
	}

	/**
	 * 根据任务id找出内存中的任务
	 * 
	 * @param ids
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected LinkedHashSet<AbstractTaskModel> findCacheTasksByIds(
			LinkedHashSet<String> ids, Jedis j) throws Exception {
		LinkedHashSet<byte[]> idBytes = Sets.newLinkedHashSet();
		for (String id : ids) {
			idBytes.add(RedisSerializeUtil.serializeString(id));
		}
		return findCacheTasks(idBytes, true, j);
	}
	
	/**
	 * 根据key模糊查询出所有的key
	 * @param key*
	 * @return 
	 * @throws Exception
	 */
	public Set<byte[]> getKeysByKey(String keys,Jedis j){
		Set<byte[]> keySets=Sets.newHashSet();
		keySets=j.keys(keys.getBytes());
		return keySets;
	}

	/**
	 * 获得指定渠道的等待队列keyByte
	 * 
	 * @param space
	 * @param channelId
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildChannelWaitQueueKeyByte(CacheDataSpace space,
			String channelId) throws Exception {
		return buildChannelQueueKeyByte(space, channelId,
				TaskCacheContext.WAIT_QUEUE, TaskCacheContext.SORTED_SET);
	}

	/**
	 * 获得指定渠道的执行队列keyByte
	 * 
	 * @param space
	 * @param channelId
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildChannelDoingQueueKeyByte(CacheDataSpace space,
			String channelId) throws Exception {
		return buildChannelQueueKeyByte(space, channelId,
				TaskCacheContext.DOING_QUEUE, TaskCacheContext.SORTED_SET);
	}
	

	/**
	 * 获得指定渠道的完成队列keyByte
	 * 
	 * @param space
	 * @param channelId
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildChannelFinishQueueKeyByte(CacheDataSpace space,
			String channelCode) throws Exception {
		return buildChannelQueueKeyByte(space, channelCode,
				TaskCacheContext.FINISH_QUEUE, TaskCacheContext.SORTED_SET);
	}
	

	/**
	 * 获取指定渠道的暂停队列keyByte
	 * 
	 * @param space
	 * @param channelId
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildChannelPauseQueueKeyByte(CacheDataSpace space,
			String channelId) throws Exception {
		return buildChannelQueueKeyByte(space, channelId,
				TaskCacheContext.PAUSE_QUEUE, TaskCacheContext.SORTED_SET);
	}

	//new
	protected byte[] buildChannelPauseQueueKeyByteNew(CacheDataSpace space,
			String channelId) throws Exception {
		return buildChannelQueueKeyByte(space, channelId,
				TaskCacheContext.PAUSE_QUEUE,TaskCacheContext.SORTED_SET);
	}
	
	/**
	 * 获得指定渠道和队列的key QUEUE:DATASPACE:queueName:channelId:queueType
	 * 
	 * @param space
	 * @param channelId
	 * @param queueName
	 * @param queueType
	 * @return
	 * @throws Exception
	 */
	protected String buildChannelQueueKey(CacheDataSpace space, String channelId,
			String queueName, String queueType) throws Exception {
		return new StringBuffer().append(TaskCacheContext.SPACE_QUEUE)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(space.getSpace())
				.append(TaskCacheContext.SPACE_DELIMITER).append(queueName)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(buildChannelIdFormatStr(channelId))
				.append(TaskCacheContext.SPACE_DELIMITER).append(queueType)
				.toString();
	}

	/**
	 * 获得指定渠道和队列的keyByte
	 * QUEUE:DATASPACE:queueName:channelId:queueType(redis的数据类型)
	 * 
	 * @param space
	 * @param channelId
	 * @param queueName
	 * @param queueType
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildChannelQueueKeyByte(CacheDataSpace space,
			String channelCode, String queueName, String queueType)
			throws Exception {
		return RedisSerializeUtil.serializeString(new StringBuffer()
				.append(TaskCacheContext.SPACE_QUEUE)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(space.getSpace())
				.append(TaskCacheContext.SPACE_DELIMITER).append(queueName)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(buildChannelIdFormatStr(channelCode))
				.append(TaskCacheContext.SPACE_DELIMITER).append(queueType)
				.toString());
	}
	
	/**
	 * 获得指定渠道统计信息的key COUNTINFO:dataspace:countInfoName:channelId:KEY
	 * 
	 * @param space
	 * @param channelId
	 * @param countInfoName
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildCountInfoKeyByte(CacheDataSpace space,
			String channelId, String countInfoName) throws Exception {
		return RedisSerializeUtil.serializeString(new StringBuffer()
				.append(TaskCacheContext.SPACE_COUNTINFO)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(space.getSpace())
				.append(TaskCacheContext.SPACE_DELIMITER).append(countInfoName)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(buildChannelIdFormatStr(channelId))
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(TaskCacheContext.KEY).toString());
	}

	/**
	 * 保存/更新 任务到缓存
	 * 
	 * @param space
	 * @param model
	 * @param pl
	 * @throws Exception
	 */
	protected void saveOrUpdateCacheTask(AbstractTaskModel m, Pipeline pl)
//	protected void saveOrUpdateCacheTask(AbstractTaskModel m,Jedis j)
			throws Exception {
		AirFreightTaskModel k=(AirFreightTaskModel)m;
//		pl.hset(buildHaskDetailKeyByte(),
//				RedisSerializeUtil.serializeString(new CacheDataSpace(m.getCreateTime(), m.getTaskType()+"",k.getTaskKey()).toString()),
//				RedisSerializeUtil.serializeObject(m));
		pl.set(new CacheDataSpace(m.getCreateTime(), m.getTaskType()+"",k.getTaskKey()).toString().getBytes(),RedisSerializeUtil.serialize(m));
//		pl.hmset(new CacheDataSpace(m.getCreateTime(), m.getTaskType()+"",k.getTaskKey()).toString(),testMap );
//		pl.sadd(buildHaskDetailKeyByte(),RedisSerializeUtil.serializeString(new CacheDataSpace(m.getCreateTime(), m.getTaskType()+"",k.getTaskKey()).toString()),RedisSerializeUtil.serializeObject(m));
	}

	/**
	 * 批量从缓存中获得任务
	 * 
	 * @param space
	 * @param idBytes
	 * @param excluseNull
	 *            排除null
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected LinkedHashSet<AbstractTaskModel> findCacheTasks(
			final LinkedHashSet<byte[]> idBytes, boolean excluseNull, Jedis j)
			throws Exception {
		Pipeline pl = openPipeline(j, false);
		byte[] hashKey = buildHaskDetailKeyByte();
		final LinkedHashSet<AbstractTaskModel> result=new LinkedHashSet<AbstractTaskModel>();
		excute(new TaskRedisCallback<LinkedHashSet<AbstractTaskModel>>() {
			@Override
			public LinkedHashSet<AbstractTaskModel> doInRedis(Jedis j)
					throws Exception {
				for (byte[] id : idBytes) {
					result.add(RedisSerializeUtil.deserializeObject(j.get(id),
							AbstractTaskModel.class));;
				}
				return result;
			}
		});
		
		/*for (byte[] id : idBytes) {
			pl.hget(hashKey,id);
		}
		List<Object> obj = syncAndReturnAllPipline(pl, false);
		LinkedHashSet<AbstractTaskModel> result = Sets.newLinkedHashSet();
		for (Object o : obj) {
			if (o == null && !excluseNull) {
				result.add(null);
			} else if (o != null) {
				result.add(RedisSerializeUtil.deserializeObject((byte[]) o,
						AbstractTaskModel.class));
			}
		}*/
		return result;
	}

	
	/**
	 * 批量从缓存中取出对应任务id的查询索引记录
	 * 
	 * @param idBytes
	 * @param excluseNull
	 *            排除null
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected LinkedHashSet<QueryIndexLog> findCacheQueryIndexLog(
			LinkedHashSet<byte[]> idBytes, boolean excluseNull, Jedis j)
			throws Exception {
		Pipeline pl = openPipeline(j, false);
		byte[] hashKey = buildQueryIndexLogHaskKeyByte();
		for (byte[] id : idBytes) {
			pl.hget(hashKey, id);
		}
		List<Object> obj = syncAndReturnAllPipline(pl, false);
		LinkedHashSet<QueryIndexLog> result = Sets.newLinkedHashSet();
		for (Object o : obj) {
			if (o == null && !excluseNull) {
				result.add(null);
			} else if (o != null) {
				result.add(RedisSerializeUtil.deserializeObject((byte[]) o,
						QueryIndexLog.class));
			}
		}
		return result;

	}

	/**
	 * 根据序列化id找出内存对应的任务，返回Map key 为Stringid,value为对应的模型对象
	 * 如果序列化id找不到对应的模型，则跳过，保证返回的map的任务均不为空
	 * 
	 * @param idBytes
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected LinkedHashMap<String, AbstractTaskModel> findCacheTasksMap(
			LinkedHashSet<byte[]> idBytes, Jedis j) throws Exception {
		Set<AbstractTaskModel> models = findCacheTasks(idBytes, true, j);
		LinkedHashMap<String, AbstractTaskModel> result = Maps
				.newLinkedHashMap();
		for (AbstractTaskModel m : models) {
			result.put(m.getId(), m);
		}
		return result;
	}
	/**
	 * 根据游离态的model找出对应的内存model
	 * @return
	 * @throws Exception
	 */
	protected LinkedHashMap<String, AbstractTaskModel> findCacheTasksMapByDissociateTasks(
			LinkedHashSet<AbstractTaskModel> dissociateModels, Jedis j)throws Exception{
		Set<AbstractTaskModel> models =findCacheTasksByDissociateTasks(dissociateModels, true, j);
		LinkedHashMap<String, AbstractTaskModel> result = Maps
				.newLinkedHashMap();
		for (AbstractTaskModel m : models) {
			result.put(m.getId(), m);
		}
		return result;
	}
	
	
	

	/**
	 * 根据序列化id找出内存对应的任务查询索引记录
	 * 
	 * @param idBytes
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected LinkedHashMap<String, QueryIndexLog> findCacheQueryIndexLogMap(
			LinkedHashSet<byte[]> idBytes, Jedis j) throws Exception {
		Set<QueryIndexLog> logs = findCacheQueryIndexLog(idBytes, true, j);
		LinkedHashMap<String, QueryIndexLog> result = Maps.newLinkedHashMap();
		for (QueryIndexLog l : logs) {
			result.put(l.getTargetTaskId(), l);
		}
		return result;
	}
	
	protected LinkedHashMap<String, QueryIndexLog> findCacheQueryIndexLogMapByDissociateTasks(
			LinkedHashSet<AbstractTaskModel> dissociateModels, Jedis j) throws Exception {
		LinkedHashSet<byte[]> idBytes=Sets.newLinkedHashSet();
		for(AbstractTaskModel m:dissociateModels){
			idBytes.add(RedisSerializeUtil.serializeString(m.getId()));
		}
		return findCacheQueryIndexLogMap(idBytes, j);
	}
	
	
	
	

	/**
	 * 过滤出已经存在于内存的任务，且返回结果集
	 * 
	 * @param space
	 * @param models
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected Set<AbstractTaskModel> filterIncludeExist(
			Set<AbstractTaskModel> models, Jedis j) throws Exception {
		Set<AbstractTaskModel> result = Sets.newLinkedHashSet();
		List<AbstractTaskModel> temp = Lists.newArrayList(models);
		List<Boolean> bls = cacheTasksExist(temp, j);
		for (int i = 0; i < bls.size(); i++) {
			if (bls.get(i) != null && bls.get(i)) {
				result.add(temp.get(i));
			}
		}
		return result;
	}

	/**
	 * 过滤出不存在内存中的任务，且返回结果集
	 * 
	 * @param space
	 * @param models
	 * @param j
	 * @return
	 * @throws Exception
	 */
	protected Set<AbstractTaskModel> filterIncludeNotExist(
			Set<AbstractTaskModel> models, Jedis j) throws Exception {
		Set<AbstractTaskModel> result = Sets.newLinkedHashSet();
		List<AbstractTaskModel> temp = Lists.newArrayList(models);
		List<Boolean> bls = cacheTasksExist(temp, j);
		for (int i = 0; i < bls.size(); i++) {
			if (bls.get(i) != null && !bls.get(i)) {
				result.add(temp.get(i));
			}
		}
		return result;
	}

	/**
	 * 任务是否已经存在
	 * 
	 * @param space
	 * @param models
	 *            有序列表
	 * @return
	 * @throws Exception
	 */
	protected List<Boolean> cacheTasksExist(List<AbstractTaskModel> models,
			Jedis j) throws Exception {
		Pipeline pl = openPipeline(j, false);
		for (AbstractTaskModel m : models) {
			pl.hexists(buildHaskDetailKeyByte(),
					RedisSerializeUtil.serializeString(m.getId()));
		}
		return syncAndReturnAllPipline(pl, false, Boolean.class, false);
	}

	/**
	 * 获得一个taskHash的key TASKDETAIL:HASH
	 * 
	 * @param space
	 * @return
	 */
	protected String buildHaskDetailKey() throws Exception {
		/*
		 * return new
		 * StringBuffer().append(TaskCacheContext.SPACE_TASKDETAIL).append
		 * (TaskCacheContext.SPACE_DELIMITER)
		 * .append(TaskCacheContext.HASH).toString();
		 */
		return "TASKDETAIL:HASH";
	}

	/**
	 * 获得一个taskHash的byteKey TASKDETAIL:dataSpace:HASH
	 * 
	 * @param space
	 * @return
	 */
	protected byte[] buildHaskDetailKeyByte() throws Exception {
		return RedisSerializeUtil.serializeString(buildHaskDetailKey());
	}

	/**
	 * 获得一个查询索引记录的haskKey
	 * 
	 * @return
	 * @throws Exception
	 */
	protected String buildQueryIndexLogHaskKey() throws Exception {
		return "QUERYINDEXLOG:HASH";
	}

	/**
	 * 获得一个查询索引记录的haskKeyByte
	 * 
	 * @return
	 * @throws Exception
	 */
	protected byte[] buildQueryIndexLogHaskKeyByte() throws Exception {
		return RedisSerializeUtil.serializeString(buildQueryIndexLogHaskKey());
	}

	/**
	 * 获取查询结果集记录的key
	 * 
	 * @param key
	 * @return
	 * @throws Exception
	 */
	protected String buildQueryResultIdKey(String key) throws Exception {
		return key + TaskCacheContext.SPACE_DELIMITER + TaskCacheContext.LIST;
	}

	/**
	 * 打开管道并且决定是否打开事务
	 * 
	 * @param j
	 * @param isMulti
	 * @return
	 * @throws Exception
	 */
	protected Pipeline openPipeline(Jedis j, boolean isMulti) throws Exception {
		Pipeline pl = j.pipelined();
		if (isMulti) {
			pl.multi();
		}
		return pl;
	}

	/**
	 * 关闭管道，不返回任务结果
	 * 
	 * @param pl
	 * @param isMulti
	 * @throws Exception
	 */
	protected void syncPipline(Pipeline pl, boolean isMulti)
			throws MultiTransactionFailException, Exception {
		Response<List<Object>> response = null;
		if (isMulti) {
			response = pl.exec();
		}
		pl.sync();
		if (response != null && response.get() == null) {
			throw new MultiTransactionFailException();
		}
	}

	/**
	 * 关闭管道，返回任务结果
	 * 
	 * @param pl
	 * @param isMulti
	 * @return
	 * @throws MultiTransactionFailException
	 * @throws Exception
	 */
	protected List<Object> syncAndReturnAllPipline(Pipeline pl, boolean isMulti)
			throws MultiTransactionFailException, Exception {
		Response<List<Object>> response = null;
		if (isMulti) {
			response = pl.exec();
		}
		List<Object> result = pl.syncAndReturnAll();
		if (response != null && response.get() == null) {
			throw new MultiTransactionFailException();
		}
		return result;
	}

	/**
	 * 关闭管道，返回任务结果,且强转类型
	 * 
	 * @param pl
	 * @param isMulti
	 * @param targetClzz
	 * @param filterNull
	 *            是否过滤空元素
	 * @return
	 * @throws MultiTransactionFailException
	 * @throws Exception
	 */
	protected <T> List<T> syncAndReturnAllPipline(Pipeline pl, boolean isMulti,
			Class<T> targetClzz, boolean excludeNull)
			throws MultiTransactionFailException, Exception {
		return transform(syncAndReturnAllPipline(pl, isMulti), targetClzz,
				excludeNull);
	}

	/**
	 * 将结果集强转为指定类型
	 * 
	 * @param list
	 * @param targetClzz
	 * @param excludeNull
	 *            排除null类型
	 * @return
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	protected <T> List<T> transform(List<Object> list, Class<T> targetClzz,
			boolean excludeNull) throws Exception {
		List<T> result = Lists.newArrayList();
		for (Object obj : list) {
			if (obj == null && excludeNull) {
				continue;
			}
			result.add((T) obj);
		}
		return result;
	}

	/**
	 * 获得配置的默认等锁超时时间
	 * 
	 * @return
	 */
	protected long getDefaultWaitLockExpire() {
		return PropertyUtils.getIntProperty(TaskServicePropertiesLoader.TASKSERVICE_CACHE_LOCK_WAIT_MAXEXPIRE,TaskServicePropertiesLoader.FILE_NAME, 20000);
	}

	/**
	 * 获得配置的默认锁超时时间
	 * 
	 * @return
	 */
	protected long getDefaultLockKeyExpire() {
		return PropertyUtils.getIntProperty(TaskServicePropertiesLoader.TASKSERVICE_CACHE_LOCK_LOCKED_MAXEXPIRE,TaskServicePropertiesLoader.FILE_NAME, 300000);
	}

	/**
	 * 执行单渠道的默认的加锁回调操作
	 * 
	 * @param d
	 * @param action
	 * @param channelIds
	 * @return
	 * @throws Exception
	 */
	protected <T> T excuteDefaultLock(TaskRedisCallback<T> action,
			CacheDataSpace d, String channelId) throws Exception {
		return getTaskRedisTemplate().executeLock(action,
				buildChannelDefaultLockKey(d, channelId),
				getDefaultWaitLockExpire(), getDefaultLockKeyExpire());
	}

	/**
	 * 执行未知的多渠道和多数据空间的加锁回调操作
	 * 
	 * @param action
	 * @param models
	 * @return
	 * @throws Exception
	 */
		TaskRedisTemplate tep=new TaskRedisTemplate();
		protected <T> T excuteDefaultLocks(TaskRedisCallback<T> action,Set<AbstractTaskModel> models) throws Exception {
//		return tep.executeLocks(action, buildChannelDefaulrLockKeys(models), getDefaultWaitLockExpire(), getDefaultLockKeyExpire());
		return getTaskRedisTemplate().executeLocks(action, buildChannelDefaulrLockKeys(models), getDefaultWaitLockExpire(), getDefaultLockKeyExpire());
	}

	/**
	 * 执行未知的多渠道和多数据空间的加锁回调操作
	 * 
	 * @param action
	 * @param models
	 * @return
	 * @throws Exception
	 */
	protected <T> T excuteDefaultLocks(TaskRedisCallback<T> action,
			AbstractTaskModel model) throws Exception {
		return getTaskRedisTemplate().executeLock(action,
				model.getId(),
				getDefaultWaitLockExpire(), getDefaultLockKeyExpire());
	}

	protected Set<String> buildChannelDefaulrLockKeys(
			Set<AbstractTaskModel> models) throws Exception {
		Set<String> keys = Sets.newLinkedHashSet();
		for (AbstractTaskModel m : models) {
			AirFreightTaskModel k=(AirFreightTaskModel)m;
			keys.add(buildChannelDefaultLockKey(new CacheDataSpace(m.getCreateTime(), m.getTaskType()), new CacheDataSpace(m.getCreateTime(), m.getTaskType()+"",k.getTaskKey()).toString()));
		}
		return keys;
	}

	/**
	 * 根据渠道生成对应的读写锁Key
	 * 
	 * @param space
	 * @param channelId
	 * @return
	 * @throws Exception
	 */
	protected String buildChannelDefaultLockKey(CacheDataSpace space,String channelId) throws Exception {
		return buildChannelKey(space, channelId, LockTypeEnum.ReadWrite);
	}

	/**
	 * 根据渠道和锁类型生成对应的锁Key
	 * 
	 * @param space
	 * @param channelId
	 * @param type
	 * @return
	 * @throws Exception
	 * LOCK:20141029000001:4f82a92c-50ad-4e24-8634-32ea7388051f:3:KEY
	 */
	protected String buildChannelKey(CacheDataSpace space, String channelId,
			LockTypeEnum type) throws Exception {
		return new StringBuffer().append(TaskCacheContext.LOCK)
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(space.getSpace())
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(buildChannelIdFormatStr(channelId))
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(type.getType())
				.append(TaskCacheContext.SPACE_DELIMITER)
				.append(TaskCacheContext.KEY).toString();
	}

	protected <T> T excute(TaskRedisCallback<T> action) throws Exception {
		return getTaskRedisTemplate().execute(action);
	}

	/**
	 * 找出任务模型的所有渠道ID
	 * 
	 * @param models
	 * @return
	 * @throws Exception
	 */
	protected Set<Long> findChannelIds(Collection<AbstractTaskModel> models)
			throws Exception {
		Set<Long> result = Sets.newLinkedHashSet();
		for (AbstractTaskModel m : models) {
			result.add(m.getChannel().getId());
		}
		return result;
	}

	/**
	 * 将渠道ID变成格式化字符串001,012,024等
	 * 
	 * @param channelId
	 * @return
	 * @throws Exception
	 */
	protected String buildChannelIdFormatStr(String channelCode) throws Exception {
		validateChannelId(channelCode);
		return StringUtils.leftPad(channelCode + "", 3, "0");
	}

	/**
	 * 保存查询索引
	 * 
	 * @param model
	 * @param pl
	 * @throws Exception
	 */
	protected QueryIndexLog saveQueryIndex(AbstractTaskModel model, Pipeline pl)
			throws Exception {
		return operateQueryIndexAndReturnLog(model, null, pl);
	}

	/**
	 * 更新查询索引
	 * 
	 * @param model
	 * @param oldLog
	 * @param pl
	 * @return
	 * @throws Exception
	 */
	protected QueryIndexLog updateQueryIndex(AbstractTaskModel model,
			QueryIndexLog oldLog, Pipeline pl) throws Exception {
		return operateQueryIndexAndReturnLog(model, oldLog == null ? null
				: oldLog.getQueryIndex(), pl);
	}

	private QueryIndexLog operateQueryIndexAndReturnLog(
			AbstractTaskModel model, QueryIndex oldIndex, Pipeline pl)
			throws Exception {
		QueryIndexLog log = null;
		TaskAddQueryIndexApplication<AbstractTaskModel> application = getAddQueryIndexApplication(model);
		if (application != null) {
			QueryIndex qi = new QueryIndex();
			application.addQueryIndex(qi, model);//redis存储目录
			if (!qi.getIndexMap().isEmpty()) {
				operateQueryIndex(model, oldIndex, qi, pl);
				// 将QueryIndexLog保存
				log = new QueryIndexLog();
				log.setQueryIndex(qi);
				log.setTargetTaskId(model.getId());
				pl.hset(buildQueryIndexLogHaskKeyByte(),
						RedisSerializeUtil.serializeString(model.getId()),
						RedisSerializeUtil.serializeObject(log));
			}
		}
		return log;
		
		/*
		 * protected void saveOrUpdateCacheTask(AbstractTaskModel model, Pipeline pl)
			throws Exception {
		pl.hset(buildHaskDetailKeyByte(),
				RedisSerializeUtil.serializeString(model.getId()),
				RedisSerializeUtil.serializeObject(model));
	}
		*/
	}

	/**
	 * 操作新旧两个索引
	 * 
	 * @param model
	 * @param oldIndex
	 * @param newIndex
	 * @param pl
	 * @throws Exception
	 */
	private void operateQueryIndex(AbstractTaskModel model,
			QueryIndex oldIndex, QueryIndex newIndex, Pipeline pl)
			throws Exception {
		CacheDataSpace space = new CacheDataSpace(model);
		if (isQueryIndexBlank(oldIndex) && isQueryIndexBlank(newIndex)) {// 两个索引都为null
			return;
		} else if (isQueryIndexBlank(oldIndex) && !isQueryIndexBlank(newIndex)) {// 只有新索引
			for (Entry<String, Object> en : newIndex.getIndexMap().entrySet()) {
				operateCondition(space, en.getKey(), en.getValue(),
						model.getId(), SAVE_OPERATE, pl);// 添加新索引
			}
		} else if (!isQueryIndexBlank(oldIndex) && isQueryIndexBlank(newIndex)) {// 没有新索引，只有老索引
			for (Entry<String, Object> en : oldIndex.getIndexMap().entrySet()) {
				operateCondition(space, en.getKey(), en.getValue(),
						model.getId(), DELETE_OPERATE, pl);// 删除老索引
			}
		} else if (!isQueryIndexBlank(oldIndex) && !isQueryIndexBlank(newIndex)) {// 新老索引皆有，最复杂
			Set<String> oldKeys = oldIndex.getIndexMap().keySet();
			for (String ok : oldKeys) {
				if (!newIndex.getIndexMap().containsKey(ok)) {// 新索引没有老索引key
					operateCondition(space, ok, oldIndex.getIndexMap().get(ok),
							model.getId(), DELETE_OPERATE, pl);
				} else if (!oldIndex.getIndexMap().get(ok)
						.equals(newIndex.getIndexMap().get(ok))) {
					// 新索引存在老索引的key且不相等
					operateCondition(space, ok, oldIndex.getIndexMap().get(ok),
							model.getId(), DELETE_OPERATE, pl);// 删除老索引
					operateCondition(space, ok, newIndex.getIndexMap().get(ok),
							model.getId(), SAVE_OPERATE, pl);// 添加新索引
				}
			}
			for (String nk : newIndex.getIndexMap().keySet()) {
				if (!oldKeys.contains(nk)) {// 找出不存在老索引中的新索引的key
					operateCondition(space, nk, newIndex.getIndexMap().get(nk),
							model.getId(), SAVE_OPERATE, pl);// 添加新索引
				}
			}
		}
	}

	/**
	 * 该索引是否null或无用
	 * 
	 * @param index
	 * @return
	 */
	private boolean isQueryIndexBlank(QueryIndex index) {
		return index == null || index.getIndexMap() == null
				|| index.getIndexMap().isEmpty();
	}

	// 操作条件值得变化
	private void operateCondition(CacheDataSpace space, String key,
			Object value, String taskId, int operate, Pipeline pl)
			throws Exception {
		if (value instanceof String) {
			switch (operate) {
			case SAVE_OPERATE:
				pl.sadd(buildConditionSetKeyByte(space, key, (String) value),
						RedisSerializeUtil.serializeString(taskId));
				break;
			case DELETE_OPERATE:
				pl.srem(buildConditionSetKeyByte(space, key, (String) value),
						RedisSerializeUtil.serializeString(taskId));
				break;
			}
		} else if (value instanceof Double) {
			switch (operate) {
			case SAVE_OPERATE:
			case UPDATE_OPERATE:
				pl.zadd(buildConditionSortSetKeyByte(space, key),
						(Double) value,
						RedisSerializeUtil.serializeString(taskId));
				break;
			case DELETE_OPERATE:
				pl.zrem(buildConditionSortSetKeyByte(space, key),
						RedisSerializeUtil.serializeString(taskId));
				break;
			}
		}
	}

}
